package com.qingyunge.app.func;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.qingyunge.bean.TableProcess;
import com.qingyunge.common.FlinkPhConfig;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Set;

public class TableProcessFunction extends BroadcastProcessFunction<JSONObject, String, JSONObject> {
    private Connection connection;
    private MapStateDescriptor<String, TableProcess> mapStateDescriptor;
    public TableProcessFunction(MapStateDescriptor<String, TableProcess> mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        connection = DriverManager.getConnection(FlinkPhConfig.PHOENIX_SERVER,FlinkPhConfig.HBASE_SCHEMA,FlinkPhConfig.PHOENIX_DRIVER);
    }

    @Override
    public void processBroadcastElement(String s, Context context, Collector<JSONObject> collector) throws Exception {
        JSONObject jsonObject = JSON.parseObject(s);
        TableProcess after = JSON.parseObject(jsonObject.getString("after"), TableProcess.class);
        //校验并建表
        checkTable(after.getSinkTable(),
                after.getSinkColumns(),
                after.getSinkPk(),
                after.getSinkExtend());
        //写入状态 广播出去
        BroadcastState<String, TableProcess> broadcastState = context.getBroadcastState(mapStateDescriptor);
        broadcastState.put(after.getSourceTable(),after);
    }

    /**
     *
     * @param sinkTable
     * @param sinkColumns
     * @param sinkPk
     * @param sinkExtend
     */
    private void checkTable(String sinkTable, String sinkColumns, String sinkPk, String sinkExtend) {
        PreparedStatement preparedStatement = null;
        // 处理特殊字段
        try {
            if(sinkPk == null || sinkPk.equals("")){
                sinkPk = "id";
            }
            if(sinkExtend == null){
                sinkExtend = "";
            }
            //sql 拼接：create table if not exists db.tn(id varchar primary key, name varchar)
            StringBuilder createTableSql = new StringBuilder("create table if not exists ")
                    .append(FlinkPhConfig.HBASE_SCHEMA)
                    .append(".")
                    .append(sinkTable)
                    .append("(");
            String[] columns = sinkColumns.split(",");
            for(int i = 0;i<columns.length;i++){
                String column = columns[i];
                if(column.equals(sinkPk)) {
                    createTableSql.append(column).append(" varchar primary key");
                }else {
                    createTableSql.append(column).append(" varchar");
                }
                if(i<columns.length-1){
                    createTableSql.append(",");
                }
            }
            createTableSql.append(")").append(sinkExtend);
            System.out.println("createTableSql = " + createTableSql);
            // 编译sql
            preparedStatement = connection.prepareStatement(createTableSql.toString());
            // 执行sql
            preparedStatement.execute();
        } catch (SQLException throwables) {
            throw new RuntimeException("建表失败： "+ sinkTable);
        } finally {
            if(preparedStatement != null){
                try {
                    preparedStatement.close();
                } catch (SQLException throwables) {
                    throwables.printStackTrace();
                }
            }
        }

    }

    @Override
    public void processElement(JSONObject jsonObject, ReadOnlyContext readOnlyContext, Collector<JSONObject> collector) throws Exception {
        //获取广播配置数据
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = readOnlyContext.getBroadcastState(mapStateDescriptor);
        String table = jsonObject.getString("table");
        TableProcess tableProcess = broadcastState.get(table);  // null ?
        if(tableProcess != null){
            // 过滤 columns 字段
            filterColumn(jsonObject.getJSONObject("data"),tableProcess.getSinkColumns());
            // 3 .补充sinktable
            jsonObject.put("sinkTable",tableProcess.getSinkTable());
            collector.collect(jsonObject);
        }else{
            System.out.println("找不到key="+table);
        }

    }

    private void filterColumn(JSONObject data, String sinkColumns) {
        String[] coluimn = sinkColumns.split(",");
        List<String> columnList = Arrays.asList(coluimn);

//        Set<Map.Entry<String, Object>> entries = data.entrySet();
//        Iterator<Map.Entry<String, Object>> iterator = entries.iterator();
//        while (iterator.hasNext()){
//            Map.Entry<String, Object> next = iterator.next();
//            if(!columnList.contains(next.getKey())){
//                iterator.remove();
//            }
//        }

        Set<Map.Entry<String, Object>> entries = data.entrySet();
        entries.removeIf(next -> !columnList.contains(next.getKey()));
    }

}
